package org.dromara.mqtt.service.impl;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mqtt.config.MqttProperties;
import org.dromara.mqtt.service.MqttMessageHandler;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Service;

import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class MqttClientService {

    @Resource
    private MqttProperties mqttProperties;

    @Resource
    private MqttMessageHandler mqttMessageHandler;

    private MqttClient mqttClient;
    private final ConcurrentHashMap<String, IMqttMessageListener> subscribers = new ConcurrentHashMap<>();

    @PostConstruct
    public void connect() throws MqttException {
        // 检查是否启用MQTT功能
        if (!mqttProperties.isEnabled()) {
            log.info("MQTT功能未启用，跳过连接");
            return;
        }

        // 检查必要配置是否存在
        if (mqttProperties.getBroker() == null || mqttProperties.getBroker().isEmpty()) {
            log.warn("MQTT Broker地址未配置，跳过连接");
            return;
        }

        if (mqttProperties.getClientId() == null || mqttProperties.getClientId().isEmpty()) {
            log.warn("MQTT ClientId未配置，跳过连接");
            return;
        }

        MemoryPersistence persistence = new MemoryPersistence();
        mqttClient = new MqttClient(mqttProperties.getBroker(), mqttProperties.getClientId(), persistence);

        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setConnectionTimeout(30);
        options.setKeepAliveInterval(60);

        if (mqttProperties.getUsername() != null && !mqttProperties.getUsername().isEmpty()) {
            options.setUserName(mqttProperties.getUsername());
        }

        if (mqttProperties.getPassword() != null && !mqttProperties.getPassword().isEmpty()) {
            options.setPassword(mqttProperties.getPassword().toCharArray());
        }

        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                log.info("MQTT连接完成: serverURI={}, reconnect={}", serverURI, reconnect);
                // 重新订阅所有主题,手动定义订阅主题
//                resubscribeAll();
//                resubscribeAllWithHandler();
            }

            @Override
            public void connectionLost(Throwable cause) {
                log.error("MQTT连接丢失,准备重新连接", cause);
                int maxRetries = 5;
                int retryCount = 0;
                long retryInterval = 5000; // 5秒

                while (true) {
                    try {
                        log.info("MQTT重连尝试 {}/{}", retryCount + 1, maxRetries);
                        mqttClient.connect(options);
                        log.info("MQTT重连成功");
                        return;
                    } catch (MqttException e) {
                        retryCount++;
                        if (retryCount >= maxRetries) {
                            log.error("MQTT重连失败，已达到最大重试次数 {}", maxRetries, e);
                            // 通知上层应用连接失败
                            handleReconnectFailure(e);
                            return;
                        }
                        log.warn("MQTT重连失败，{}秒后进行第{}次重试", retryInterval / 1000, retryCount + 1, e);
                        try {
                            Thread.sleep(retryInterval);
                        } catch (InterruptedException ie) {
                            Thread.currentThread().interrupt();
                            log.warn("重连等待被中断");
                            return;
                        }
                        retryInterval = Math.min(retryInterval * 2, 30000); // 指数退避，最大30秒
                    }
                }
            }

            private void handleReconnectFailure(MqttException e) {
                // 可以在这里通知上层应用或执行其他失败处理逻辑
            }


            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                // 此方法不会被调用，因为使用了setCallbackMultiThreaded
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                log.debug("消息发送完成: messageId={}", token.getMessageId());
            }
        });

        mqttClient.connect(options);
        log.info("MQTT客户端连接成功: broker={}", mqttProperties.getBroker());
    }

    @PreDestroy
    public void disconnect() throws MqttException {
        if (mqttClient != null && mqttClient.isConnected()) {
            mqttClient.disconnect();
            mqttClient.close();
            log.info("MQTT客户端已断开连接");
        }
    }

    /**
     * 发布消息
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量等级
     * @param retained 是否保留消息
     */
    public void publish(String topic, String payload, int qos, boolean retained) throws MqttException {
        if (!mqttClient.isConnected()) {
            log.warn("MQTT客户端未连接，无法发布消息");
            return;
        }

        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        message.setRetained(retained);

        mqttClient.publish(topic, message);
        log.debug("发布消息: topic={}, payload={}", topic, payload);
    }

    /**
     * 发布消息（使用默认QoS和retained设置）
     * @param topic 主题
     * @param payload 消息内容
     */
    public void publish(String topic, String payload) throws MqttException {
        publish(topic, payload, 1, false);
    }

    /**
     * 订阅主题
     * @param topic 主题
     * @param qos 服务质量等级
     * @param listener 消息监听器
     */
    public void subscribe(String topic, int qos, IMqttMessageListener listener) throws MqttException {
        if (!mqttClient.isConnected()) {
            log.warn("MQTT客户端未连接，无法订阅主题: {}", topic);
            return;
        }

        mqttClient.subscribe(topic, qos, listener);
        subscribers.put(topic, listener);
        log.info("订阅主题: topic={}, qos={}", topic, qos);
    }

    /**
     * 订阅主题（使用默认QoS）
     * @param topic 主题
     * @param listener 消息监听器
     */
    public void subscribe(String topic, IMqttMessageListener listener) throws MqttException {
        subscribe(topic, 1, listener);
    }

    /**
     * 订阅主题并使用自定义消息处理器
     *  给App端调用指定的主题，后端会记录日志
     * @param topic 主题
     * @param qos 服务质量等级
     */
    public void subscribeWithHandler(String topic, int qos) throws MqttException {
        IMqttMessageListener listener = (t, msg) -> {
            String payload = new String(msg.getPayload());
            log.info("收到消息: topic={}, message={}", t, payload);
            // 使用注册的消息处理器处理消息
            if (mqttMessageHandler != null) {
                mqttMessageHandler.handleMessage(t, payload);
            }
        };

        subscribe(topic, qos, listener);
    }

    /**
     * 重新订阅所有主题，带消息处理器
     */
    private void resubscribeAllWithHandler() {
        try {
            for (String topic : subscribers.keySet()) {
                subscribeWithHandler(topic, 1);
                log.info("重新订阅带消息处理器主题: {}", topic);
            }
        } catch (MqttException e) {
            log.error("重新订阅带消息处理器主题失败", e);
        }
    }

    /**
     * 取消订阅
     * @param topic 主题
     */
    public void unsubscribe(String topic) throws MqttException {
        if (!mqttClient.isConnected()) {
            log.warn("MQTT客户端未连接，无法取消订阅主题: {}", topic);
            return;
        }

        mqttClient.unsubscribe(topic);
        subscribers.remove(topic);
        log.info("取消订阅主题: {}", topic);
    }

    /**
     * 重新订阅所有主题
     */
    private void resubscribeAll() {
        try {
            for (String topic : subscribers.keySet()) {
                IMqttMessageListener listener = subscribers.get(topic);
                mqttClient.subscribe(topic, listener);
                log.info("重新订阅主题: {}", topic);
            }
        } catch (MqttException e) {
            log.error("重新订阅失败", e);
        }
    }

    /**
     * 检查是否已连接
     * @return 是否连接
     */
    public boolean isConnected() {
        return mqttClient != null && mqttClient.isConnected();
    }
}
